package com.hlz.flink.chapter06.window;

import com.hlz.flink.chapter05.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * 处理迟到的数据
 *
 * @author Hongliang Zhu
 * @create 2023-06-23 21:13
 */
public class LateDataTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<Event> singleOutputStreamOperator = env.socketTextStream("localhost", 9696)
                .map(data -> {
                    String[] fields = data.split(",");
                    return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
                });

        // 设置watermark，延迟2s
        SingleOutputStreamOperator<Event> eventSingleOutputStreamOperator = singleOutputStreamOperator.assignTimestampsAndWatermarks(WatermarkStrategy.
                <Event>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
            @Override
            public long extractTimestamp(Event element, long recordTimestamp) {
                return element.getTimeStamp();
            }
        }));

        // 定义侧输出流标签
        OutputTag<Event> outputTag = new OutputTag<Event>("late") {
        };

        SingleOutputStreamOperator<UrlViewCount> result = eventSingleOutputStreamOperator.keyBy(data -> data.getUrl())
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                // 允许窗口处理迟到的数据吗，设置一分钟的等待时间
                .allowedLateness(Time.minutes(1))
                .sideOutputLateData(outputTag)
                .aggregate(new UrlViewCountAgg(), new UrlViewCountResult());

        result.print("result");

        DataStream<Event> sideOutput = result.getSideOutput(outputTag);

        sideOutput.print("side-output");


        // 为方便观察，可以将原始数据也输出
        singleOutputStreamOperator.print("input");
        env.execute();
    }
}

/**
 nc -lk 9696
输入：

 Alice, ./home, 1000
 Alice, ./home, 2000
 Alice, ./home, 10000
 Alice, ./home, 9000
 Alice, ./cart, 12000
 Alice, ./prod?id=100, 15000
 Alice, ./prod?id=100, 22000
 Alice, ./home, 9000
 Alice, ./home, 8000
 Alice, ./prod?id=200, 70000
 Alice, ./home, 8000
 Alice, ./prod?id=300, 72000
 lice, ./home, 8000

 输出：
 input> Event{user='Alice', url='./home', timeStamp=1970-01-01 08:00:01.0}
 input> Event{user='Alice', url='./home', timeStamp=1970-01-01 08:00:02.0}
 input> Event{user='Alice', url='./home', timeStamp=1970-01-01 08:00:10.0}
 input> Event{user='Alice', url='./home', timeStamp=1970-01-01 08:00:09.0}
 input> Event{user='Alice', url='./cart', timeStamp=1970-01-01 08:00:12.0}
 result> UrlViewCount{url='./home', count=3, windowStart=0, windowEnd=10000}
 input> Event{user='Alice', url='./prod?id=100', timeStamp=1970-01-01 08:00:15.0}
 input> Event{user='Alice', url='./prod?id=100', timeStamp=1970-01-01 08:00:22.0}
 result> UrlViewCount{url='./cart', count=1, windowStart=10000, windowEnd=20000}
 result> UrlViewCount{url='./home', count=1, windowStart=10000, windowEnd=20000}
 result> UrlViewCount{url='./prod?id=100', count=1, windowStart=10000, windowEnd=20000}
 input> Event{user='Alice', url='./home', timeStamp=1970-01-01 08:00:09.0}
 result> UrlViewCount{url='./home', count=4, windowStart=0, windowEnd=10000}
 input> Event{user='Alice', url='./home', timeStamp=1970-01-01 08:00:08.0}
 result> UrlViewCount{url='./home', count=5, windowStart=0, windowEnd=10000}
 input> Event{user='Alice', url='./prod?id=200', timeStamp=1970-01-01 08:01:10.0}
 result> UrlViewCount{url='./prod?id=100', count=1, windowStart=20000, windowEnd=30000}
 input> Event{user='Alice', url='./home', timeStamp=1970-01-01 08:00:08.0}
 result> UrlViewCount{url='./home', count=6, windowStart=0, windowEnd=10000}
 input> Event{user='Alice', url='./prod?id=300', timeStamp=1970-01-01 08:01:12.0}
 input> Event{user='lice', url='./home', timeStamp=1970-01-01 08:00:08.0}
 side-output> Event{user='lice', url='./home', timeStamp=1970-01-01 08:00:08.0}
 result> UrlViewCount{url='./prod?id=300', count=1, windowStart=70000, windowEnd=80000}
 result> UrlViewCount{url='./prod?id=200', count=1, windowStart=70000, windowEnd=80000}
 */
